package momo.app.status;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.Iterator;
import java.util.StringTokenizer;

import momo.app.comms.CommsManager;
import momo.app.comms.Delayer;
import momo.app.comms.StateResetter;
import momo.app.config.AppConfig;
import momo.app.structure.Message;
import momo.app.structure.Peer;
import momo.app.structure.PeerLink;
import momo.app.structure.SessionController;
import momo.app.structure.SessionTracker;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class StatusHandler extends Thread
{
	private final int TIMEOUT = 2 * 60 * 1000; 
	private Log log = LogFactory.getLog(StatusHandler.class);
	
	private Socket sock;
	private InetAddress connecter;
	private BufferedReader in;
	private PrintWriter out;
	private RcvMsgHistory history;
	private StatusManager statusManager;
	private CommsManager commsManager;
	
	public StatusHandler(Socket sock) throws IOException
	{
		this.sock = sock;
		sock.setSoTimeout(TIMEOUT);
		connecter = sock.getInetAddress();
		in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
		out = new PrintWriter(sock.getOutputStream(), true);
		history = RcvMsgHistory.getInstance();
		statusManager = StatusManager.getInstance();
		commsManager = CommsManager.getInstance();
		this.start();
	}
	
	public void run()
	{
		if ( log.isDebugEnabled() )
			log.debug("StatusHandler starts for " + sock.getInetAddress().getHostAddress());
		
		try
		{
			String data = in.readLine();
			
			if ( data != null )
			{
				Message msg = Message.createMessage(data);
				if ( msg != null )
				{
					handle(msg);
				}else
					throw new UnsupportedEncodingException();
			}			
		}catch(IOException ioe)
		{
			if ( log.isErrorEnabled() )
				log.error("IOException", ioe);
		}catch(Exception e)
		{
			if ( log.isErrorEnabled() )
				log.error("Exception", e);
		}
		
		try
		{
			in.close();
			out.close();
			sock.close();
		}catch(IOException ioe)
		{
			if ( log.isErrorEnabled() )
				log.error("IOException", ioe);
		}
		
		if ( log.isDebugEnabled() )
			log.debug("StatusHandler ends for " + sock.getInetAddress().getHostAddress());
	}
	
	private boolean hasRcvedBefore(Message msg)
	{
		return history.contains(msg);
	}
	
	private void storeMsg(Message msg)
	{
		history.store(msg);
	}
	
	private void handleBroadcast(Message msg)
	{
		if ( msg.getMsgType().equals(Message.NEW_PEER_LINKUP) )
			replyNewPeerLinkup(msg);
//		else if ( msg.getMsgType().equals(Message.UNKIND_LEAVE) )
//			replyUnkindLeave(msg);
		else if ( msg.getMsgType().equals(Message.GENERAL_ELECTION) )
			replyGeneralElection(msg);
		else if ( msg.getMsgType().equals(Message.NEW_STRUCTURE) )
			replyNewStructure(msg);
		else if ( msg.getMsgType().equals(Message.RESTRUCTURE_REQUIRED) )
			replyRestructureRequired(msg);
		else if ( msg.getMsgType().equals(Message.RESET_STATE) )
			replyResetState(msg);
		
	}
	
	private void handle(Message msg) throws IOException
	{
		if ( SessionController.getInstance().isSameSession(msg.getSessionId()) )
		{
			if ( msg.getForwardType().equals(Message.BROADCAST) )
			{
				if ( !hasRcvedBefore(msg) )
				{
					storeMsg(msg);
					handleBroadcast(msg);
					statusManager.broadcast(msg);
				}else {
					if ( log.isDebugEnabled() )
						log.debug("IGNORED: " + msg.toString());
				}
			}else if ( msg.getForwardType().equals(Message.NO_FORWARD) )
			{
				if ( msg.getMsgType().equals(Message.GET_PEERS) )
					replyGetPeers(msg);
				else if ( msg.getMsgType().equals(Message.LATENCY_MEASURE) )
					replyLatMeasure(msg);
				else if ( msg.getMsgType().equals(Message.STAB_MAP_REQUEST) )
					replyStabMapRequest(msg);
//				else if ( msg.getMsgType().equals(Message.JOIN) )
//					replyJoin(msg);
				
			}
		}else
		{
			if ( msg.getMsgType().equals(Message.JOIN) )
				replyJoin(msg);
			else
			{
				int mySID = SessionController.getInstance().getSessionId();
				int msgSID = msg.getSessionId();
				if ( log.isDebugEnabled() )
					log.debug(">>>>>>>>>> Diff Session! Mine: " + mySID +", Msg: " + msgSID + " <<<<<<<<<<<<<<<<");
			}
		}
	}
	
	private void replyJoin(Message msg) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug("Replying Join, MsgId: " + msg.getMsgId());
		
		Peer me = AppConfig.getInstance().getMe();
		
		StateResetter stateReset = StateResetter.getInstance();
		if ( stateReset.hasTask() )
		{
			Message msgToSend = new Message(Message.NO_FORWARD, 
					me.getId(), 
					Message.RESET_STATE_IN_PROGRESS,
					me.toString()
					);
			out.println(msgToSend.toString());
			out.flush();
		}else
		{
			Peer peer = Peer.createPeer(msg.getMsg());
			
			long start = System.currentTimeMillis();
			Message msgToSend = new Message(Message.NO_FORWARD, 
					me.getId(), 
					Message.JOIN_REPLY,
					me.toString()+"|"+statusManager.getRootId()+"|"+SessionController.getInstance().getSessionId()
					);
			out.println(msgToSend.toString());
			out.flush();
			
			in.readLine();
			long end = System.currentTimeMillis();
			double lat = LatencyMeasurement.calculateMeasuredLatency(start, end);
			
			double costFromChildToMe = peer.getUserCost(AppConfig.getInstance().getId());
			double costFrmMeToChild = AppConfig.getInstance().getUserCost(peer.getId());
			double cost = costFromChildToMe + costFrmMeToChild;
			
			PeerLink link = new PeerLink(peer.getId()+":"+me.getId(), peer, me, cost, lat);
//			CommsManager.getInstance().addPendingJoinChild(peer, link);
			CommsManager.getInstance().pendingPeer(peer, link, null);
		}
	}
	
	private void replyGetPeers(Message msg) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug("Replying Get Peers Info, MsgId: " + msg.getMsgId());		StringBuffer buffer = new StringBuffer();
		
		Iterator<Peer> iter = statusManager.getPeers().iterator();
		while ( iter.hasNext() )
			buffer.append(iter.next().toString()+"|");
		
		out.println(
				new Message(Message.NO_FORWARD,
						AppConfig.getInstance().getId(),
						Message.GET_PEERS_REPLY,
						buffer.toString()
						).toString()
				);
		out.flush();
	}
	
	private void replyLatMeasure(Message msg) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug("Replying Lat Measure, MsgId: " + msg.getMsgId());
		
		out.println(
				new Message(
						Message.NO_FORWARD, 
						AppConfig.getInstance().getId(), 
						Message.LATENCY_MEASURE_REPLY, 
						"").toString()
				);
		out.flush();
	}
	
	private void replyNewPeerLinkup(Message msg)
	{
		try
		{
			if ( log.isDebugEnabled() )
				log.debug("NEW_PEER_LINKUP: " + msg.getMsg());
			
			StringTokenizer tokens = new StringTokenizer(msg.getMsg(), "|");
			Peer peer = Peer.createPeer(tokens.nextToken());
			
			StringTokenizer stLink = new StringTokenizer(tokens.nextToken(), ",");
			String source_id = stLink.nextToken();
			String dest_id = stLink.nextToken();		
			double cost = Double.parseDouble(stLink.nextToken());
			double lat = Double.parseDouble(stLink.nextToken());
			
			if ( !peer.getId().equals(AppConfig.getInstance().getId()) )
			{
				statusManager.addPeer(peer, "replyNewPeerLinkup");
				statusManager.calWorkload("replyNewPeerLinkup");
				statusManager.noteTimeJoined(peer.getId());
				statusManager.timeResetExistingPeers();
				statusManager.displayStabMap("replyNewPeerLinkup");
				statusManager.addLatMeasureThread(peer);
				statusManager.addPeerLink(source_id, dest_id, cost, lat);
			}
			
			//restructure when new join (for planetlab)
//			Delayer.getInstance().delayMsgSent();
//			commsManager.startRestructure();
			
//			StatusManager.getInstance().displayPeer();
//			StatusManager.getInstance().displayPeerLink();
		}catch (Exception e) {
			if ( log.isErrorEnabled() )
				log.error("Exception", e);
		}
	}
	
	private void replyGeneralElection(Message msg)
	{
		//delayed all the msg by source
		if ( log.isDebugEnabled() )
			log.debug("replyGeneralElection, MsgId: " +msg.getMsgId());
		
		Delayer.getInstance().delayMsgSent();
		
		commsManager.stopElectionTimeout();
		commsManager.startElection(Message.GENERAL_ELECTION);
		StringTokenizer tokens = new StringTokenizer(msg.getMsg(), "|");
		Peer elector = Peer.createPeer(tokens.nextToken());
		int candidateId = Integer.parseInt(tokens.nextToken());
		commsManager.vote(candidateId);
	}
	
	private void replyNewStructure(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug("replyNewStructure, MsgId: " +msg.getMsgId());
		commsManager.restructure(msg);
	}
	
	private void replyRestructureRequired(Message msg)
	{
		String id = msg.getMsg();
		if ( log.isDebugEnabled() )
			log.debug("replyRestructureRequired, MsgId: " +msg.getMsgId()+", Leaver id: " + id);
		if ( statusManager.isPeerExist(id) )
		{
			//delayed all the msg by source
			Delayer.getInstance().delayMsgSent();
			
			statusManager.removeAssociationOnPeer(id);
			//its being removed cos now we are going to assume that only rcvchannel will have the ability
//			commsManager.removeAssociationOnPeer(id);
			commsManager.restructureRequired(msg);
		}else {
			if ( log.isDebugEnabled() )
				log.debug("Leaver wasnt in peer list");
		}
	}
	
	private void replyResetState(Message msg)
	{
		if ( log.isDebugEnabled() )
			log.debug("replyResetState, MsgId: " +msg.getMsgId());
		int newSID = Integer.parseInt(msg.getMsg());
		StateResetter.getInstance().reset(newSID, false);
	}
	
	private void replyStabMapRequest(Message msg)
	{
		statusManager.displayStabMap("replyStabMapRequest, MsgId: " + msg.getMsgId());
		out.println(
				new Message(Message.NO_FORWARD,
						AppConfig.getInstance().getId(),
						Message.STAB_MAP_REPLY,
						statusManager.getStabMap()
						).toString()
				);
		out.flush();
	}
	
}//end of class StatusHandler